package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

/**
 * @author brianxia
 * @version 1.0
 * @date 2022-09-27 14:20
 */
@Configuration
public class ArticleBehaviorStream {

    @Bean
    public KStream<String,String> stream(StreamsBuilder builder){
        //1.数据来源
        KStream<String, String> stream = builder.stream(HotArticleConstants.HOT_ARTICLE_BEHAVIOR_KAFKA_TOPIC);

        //2.流式处理
        //分组，首先得把数据处理一下，将文章ID取出来放到key
        stream.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                UpdateArticleMess updateArticleMess = JSON.parseObject(value, UpdateArticleMess.class);
                Long articleId = updateArticleMess.getArticleId();
                return new KeyValue<>(String.valueOf(articleId),value);
            }
        }).groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                .aggregate(new Initializer<String>() {
                    @Override
                    public String apply() {
                        //初始化4个值为0 文章ID还没给值
                        return JSON.toJSONString(new ArticleVisitStreamMess());
                    }
                }, new Aggregator<String, String, String>() {
                    @Override
                    public String apply(String key, String value, String aggregate) {
                        //key是消息key 文章id
                        //value 是 UpdateArticleMess对象JSON字符串
                        UpdateArticleMess updateArticleMess =
                                JSON.parseObject(value, UpdateArticleMess.class);
                        //aggregate ArticleVisitStreamMess对象的json字符串
                        ArticleVisitStreamMess mess = JSON.parseObject(aggregate, ArticleVisitStreamMess.class);

                        //1.将mess文章ID赋值一下
                        if(mess.getArticleId() == null){
                            mess.setArticleId(updateArticleMess.getArticleId());
                        }
                        //2.处理累加过程
                        switch (updateArticleMess.getType()){
                            case COMMENT:
                                mess.setComment(mess.getComment() + updateArticleMess.getAdd());
                                break;
                            case VIEWS:
                                mess.setView(mess.getView() + updateArticleMess.getAdd());
                                break;
                            case LIKES:
                                mess.setLike(mess.getLike() + updateArticleMess.getAdd());
                                break;
                            case COLLECTION:
                                mess.setCollect(mess.getCollect() + updateArticleMess.getAdd());
                                break;
                            default:
                                break;
                        }
                        //return ArticleVisitStreamMess对象的json字符串
                        return JSON.toJSONString(mess);
                    }
                }).toStream()
                .map(new KeyValueMapper<Windowed<String>, String, KeyValue<?, ?>>() {
                    @Override
                    public KeyValue<?, ?> apply(Windowed<String> key, String value) {
                        return new KeyValue<>(key.key(),value);
                    }
                }).to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);


        return stream;
    }
}
